package com.ls.fw.data.hbase.support;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ServerLoad;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;

import com.ls.fw.data.hbase.bean.HbaseStatus;
import com.ls.fw.data.hbase.client.HBaseClient;
import com.ls.fw.data.hbase.exception.CatcherRaiseException;
import com.ls.fw.data.hbase.exception.DuplicateException;
import com.ls.fw.data.hbase.exception.UnHandledException;
import com.ls.fw.data.hbase.exception.YHBaseException;
import com.ls.fw.data.hbase.utils.DateTimeUtil;
import com.ls.fw.data.hbase.utils.HostUtil;

/**
 * 
 * @author ls
 */
public class HBaseAdmin {
	private Admin admin;

	public HBaseAdmin(Admin admin) {
		this.admin = admin;
	}

	public boolean tableExists(TableName tableName) {
		try {
			return admin().tableExists(tableName.get());
		} catch (IOException e) {
			return false;
		}
	}

	private void checkTableNotExist(TableName tableName) {
		CATCHER(() -> {
			try {
				if (admin().tableExists(tableName.get())) {
					throw new DuplicateException("table name existed");
				}
			} catch (Throwable throwable) {
				throw new UnHandledException(throwable);
			}
		});
	}

	public Admin admin() {
		return admin;
	}

	private final void CATCHER(Runnable task) {
		try {
			task.run();
		} catch (YHBaseException d) {
			throw d;
		} catch (Throwable t) {
			throw new CatcherRaiseException(t);
		}
	}

	public void snapshot(String snapshotName, TableName tableName) {
		try {
			admin().snapshot(snapshotName, tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void restoreSnapshot(String snapshotName) {
		try {
			admin().restoreSnapshot(snapshotName);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public boolean isAborted() {
		try {
			return admin().isAborted();
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}
	public void close() {
		try {
			admin().close();
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void cloneSnapshot(String snapshotName, TableName tableName) {
		try {
			admin().cloneSnapshot(snapshotName, tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

//	public List<HBaseProtos.SnapshotDescription> listSnapshots() {
//		try {
//			return admin().listSnapshots();
//		} catch (Throwable throwable) {
//			throw new UnHandledException(throwable);
//		}
//	}

	public void deleteSnapshot(String snapshotName) {
		try {
			admin().deleteSnapshot(snapshotName);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public HTableDescriptor tableDescriptor(TableName tableName) {
		try {
			return admin().getTableDescriptor(tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void updateTable(TableName tableName, HTableDescriptor tableDescriptor) {
		try {
			admin().modifyTable(tableName.get(), tableDescriptor);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public HTableDescriptor[] listTables() {
		try {
			return admin().listTables();
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void enableTable(TableName tableName) {
		if (!isTableDisabled(tableName)) {
			return;
		}

		try {
			admin().enableTable(tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public boolean isTableDisabled(TableName tableName) {
		try {
			return admin().isTableDisabled(tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void truncateTable(TableName tableName) {
		truncateTable(tableName, false);
	}

	public void truncateTable(TableName tableName, boolean preserveSplit) {
		disableTable(tableName);
		try {
			admin().truncateTable(tableName.get(), preserveSplit);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void disableTable(TableName tableName) {
		if (!isTableDisabled(tableName)) {
			try {
				admin().disableTable(tableName.get());
			} catch (Throwable throwable) {
				throw new UnHandledException(throwable);
			}
		}
	}

	public void deleteTable(TableName tableName) {
		disableTable(tableName);
		try {
			admin().deleteTable(tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void updateCompression(TableName tableName, String family, Algorithm compression) {
		try {
			admin().modifyColumn(tableName.get(), new HColumnDescriptor(family).setCompressionType(compression));
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void addColumnFamily(TableName tableName, String family, Algorithm compression) {
		try {
			admin().addColumn(tableName.get(), new HColumnDescriptor(family).setCompressionType(compression));
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void deleteColumnFamily(TableName tableName, String family) {
		try {
			admin().deleteColumn(tableName.get(), HBaseClient.bytes(family));
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void majorCompact(TableName tableName) {
		majorCompact(tableName, null);
	}

	public void majorCompact(TableName tableName, String family) {
		try {
			admin().majorCompact(tableName.get(), family == null ? null : HBaseClient.bytes(family));
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void split(TableName tableName, byte[] splitPoint) {
		try {
			admin().split(tableName.get(), splitPoint);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public boolean balancer() {
		try {
			return admin().balancer();
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void createTable(TableName tableName) {
//		checkTableNotExist(tableName);
		try {
			admin().createTable(newTableDescriptor(tableName));
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void createTable(TableName tableName,String family, Algorithm compression) {
//		checkTableNotExist(tableName);
		try {
			HTableDescriptor t = newTableDescriptor(tableName);
			t.addFamily(new HColumnDescriptor(family).setCompressionType(compression));
			admin().createTable(t);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}
	private HTableDescriptor newTableDescriptor(TableName tableName) {
		try {
			return new HTableDescriptor(tableName.get());
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}

	public void createTable(TableName tableName, byte[] startKey, byte[] endKey, int numberOfRegions) {
		checkTableNotExist(tableName);

		try {
			admin().createTable(newTableDescriptor(tableName), startKey, endKey, numberOfRegions);
		} catch (Throwable throwable) {
			throw new UnHandledException(throwable);
		}
	}
	

	public List<HbaseStatus> getMasterStatus() throws MasterNotRunningException, ZooKeeperConnectionException, IOException {
		List<HbaseStatus> list = new ArrayList<HbaseStatus>();
		//System.out.println("getMasterStatus hbase.master："+hbaseTableManager.getConfiguration().get("hbase.master"));
		Admin hBaseAdmin = null;
		hBaseAdmin = admin();
		//master
		//System.out.println("isAborted:"+hBaseAdmin.isAborted());
		//System.out.println("isMasterRunning:"+hBaseAdmin.isMasterRunning());
		ServerName ma = hBaseAdmin.getClusterStatus().getMaster();
		ServerLoad serverLoad = hBaseAdmin.getClusterStatus().getLoad(ma);
		//hBaseAdmin.isMasterRunning()
		if (serverLoad != null) {
			 long startcode = ma.getStartcode();
				 list.add(new HbaseStatus(ma.getHostname(), HostUtil
						.getIPByHostname(ma.getHostname())
						+ ":"
						+ ma.getPort(), "master", serverLoad.getUsedHeapMB(),
						serverLoad.getMaxHeapMB(), serverLoad.getMemstoreSizeInMB(), false ,
						DateTimeUtil.getFormatDate(startcode))); 
		} else {
			 long startcode = ma.getStartcode();
			 list.add(new HbaseStatus(ma.getHostname(), HostUtil.getIPByHostname(ma
						.getHostname()) + ":" + ma.getPort(), "master",DateTimeUtil.getFormatDate(startcode), false));
		}
		Collection<ServerName> v1 = hBaseAdmin.getClusterStatus().getBackupMasters();
		if(v1!=null){
			for (ServerName serverName : v1) {
				ServerLoad sl = hBaseAdmin.getClusterStatus().getLoad(serverName);
				if (sl != null) {
					long startcode = serverName.getStartcode();
					list.add(new HbaseStatus(serverName.getHostname(), HostUtil
							.getIPByHostname(serverName.getHostname())
							+ ":"
							+ serverName.getPort(), "backupMaster", sl.getUsedHeapMB(),
							sl.getMaxHeapMB(), sl.getMemstoreSizeInMB(), false,
							DateTimeUtil.getFormatDate(startcode)));
				}else {
					long startcode = serverName.getStartcode();
					list.add(new HbaseStatus(serverName
							.getHostname(), HostUtil.getIPByHostname(serverName
							.getHostname()) + ":" + serverName.getPort(), "backupMaster",DateTimeUtil.getFormatDate(startcode),true));
				}
			}
		}
		return list;
	}


	public List<HbaseStatus> getRegionServerStatus() throws IOException {
		List<HbaseStatus> list = new ArrayList<HbaseStatus>();
		Admin hBaseAdmin =  admin();
		//当前活着的节点
		Collection<ServerName> v1 = hBaseAdmin.getClusterStatus().getServers();
		for (ServerName serverName : v1) {
			ServerLoad sl = hBaseAdmin.getClusterStatus().getLoad(serverName);
			if (sl != null) {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName.getHostname(), HostUtil
						.getIPByHostname(serverName.getHostname())
						+ ":"
						+ serverName.getPort(), "regionServer", sl.getUsedHeapMB(),
						sl.getMaxHeapMB(), sl.getMemstoreSizeInMB(), false,
						DateTimeUtil.getFormatDate(startcode)));
			}else {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName
						.getHostname(), HostUtil.getIPByHostname(serverName
						.getHostname()) + ":" + serverName.getPort(), "regionServer",DateTimeUtil.getFormatDate(startcode),true));
			}
		}
		
		//当前已经停止的节点
		Collection<ServerName> v = hBaseAdmin.getClusterStatus()
				.getDeadServerNames();
		for (ServerName serverName : v) {
			ServerLoad sl = hBaseAdmin.getClusterStatus().getLoad(serverName);
			if (sl != null) {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName.getHostname(), HostUtil
						.getIPByHostname(serverName.getHostname())
						+ ":"
						+ serverName.getPort(), "regionServer", sl.getUsedHeapMB(),
						sl.getMaxHeapMB(), sl.getMemstoreSizeInMB(), true,
						DateTimeUtil.getFormatDate(startcode)));
			} else {
				long startcode = serverName.getStartcode();
				list.add(new HbaseStatus(serverName
						.getHostname(), HostUtil.getIPByHostname(serverName
						.getHostname()) + ":" + serverName.getPort(), "regionServer",DateTimeUtil.getFormatDate(startcode),true));
			}
		}
		return list;
	}
}
